package com.shu.bolt;

import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

public class WordCountBolt extends BaseBasicBolt{

   Map<String, Integer> counts = new HashMap<String, Integer>();
   Set<String> keyName = new HashSet<String>();
       
	public void execute(Tuple input, BasicOutputCollector collector) {
		List<Object> values = input.getValues();
		Object o = values.get(0);
		List<String> words =(List<String>)o ; // 每次统计一个分词
		for(String word : words) {
		    Integer count = counts.get(word);
	        if (count == null)
	            count = 0;
	        count++;
	        counts.put(word, count);
	        collector.emit(new Values(word, count));		
		}
    
        System.out.println("counts:"+counts);
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		 declarer.declare(new Fields("word", "count"));
	}

}
